#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/protobuf/codec.h>
#include <muduo/protobuf/dispatcher.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Mutex.h>
#include <mqcomm/threadpool.hpp>
#include <mqcomm/Message.pb.h>
#include <mqcomm/protocol.pb.h>
#include "connection.hpp"
#include "consumer.hpp"
#include "virtualhost.hpp"
#include <iostream>
#include <string>
#include <vector>
using namespace std;

namespace rbmq {


class Broker {
public :
    #define DBFILE "/meta.db"
    typedef shared_ptr<google::protobuf::Message> MessagePtr; 
    Broker(string& basedir, const string& host_name, int port)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Broker", muduo::net::TcpServer::kReusePort)
        ,_dispatcher(std::bind(&Broker::OnUnkownMessage, this, placeholders:: _1, placeholders:: _2, placeholders:: _3))
        ,_codec(make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,\
         placeholders:: _1, placeholders:: _2, placeholders:: _3)))
        ,_virtualHost(make_shared<VirtualHost>(host_name, basedir, basedir + DBFILE))
        ,_consumer_manager(make_shared<consumerManager>())
        ,_connection_manager(make_shared<ConnectionManager>())
        ,_threadpool(make_shared<threadpool>())

    {
        // 注册请求处理函数
        _dispatcher.registerMessageCallback<mqNs::declareExchangeRequest>(std::bind(&Broker::onDeclareExchange, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));
        _dispatcher.registerMessageCallback<mqNs::deleteExchangeRequest>(std::bind(&Broker::onDeleteExchange, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));
        _dispatcher.registerMessageCallback<mqNs::openChannelRequest>(std::bind(&Broker::onOpenChannel, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));
        _dispatcher.registerMessageCallback<mqNs::closeChannelRequest>(std::bind(&Broker::onCloseChannel, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::declareQueueRequest>(std::bind(&Broker::onDelareQueue, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));     
        _dispatcher.registerMessageCallback<mqNs::deleteQueueRequest>(std::bind(&Broker::onDeleteQueue, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::QueueBindingRequest>(std::bind(&Broker::onQueueBinding, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::QueueUnBindingRequest>(std::bind(&Broker::onQueueUnBinding, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::basicPublshRequest>(std::bind(&Broker::onBasicPublish, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::basicAckRequest>(std::bind(&Broker::onBasicAck, this,\
         placeholders::_1, placeholders::_2, placeholders::_3));  
        _dispatcher.registerMessageCallback<mqNs::basicConsumerRequest>(std::bind(&Broker::onBasicConsumer, this,\
         placeholders::_1, placeholders::_2, placeholders::_3)); 
        _dispatcher.registerMessageCallback<mqNs::basicConcelRequest>(std::bind(&Broker::onBasicConcel, this,\
         placeholders::_1, placeholders::_2, placeholders::_3)); 

        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), placeholders::_1, placeholders::_2, placeholders::_3));
        _server.setConnectionCallback(std::bind(&Broker::OnConnection, this, placeholders::_1));
        // 需要针对历史消息中的所有队列，还要初始化队列的消费者管理句柄
        QueueMap map = _virtualHost->getAllQueue();
        // 获取所有的消费队列信息，使用消费者管理句柄完成初始化
        for (auto& p : map) {
            _consumer_manager->InitQueueConsumer(p.first);
        }
    }


    void Start() {
        _server.start();
        _baseloop.loop();
    }

private :
    // 一大批请求处理函数
    // 打开信道
    void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& req, muduo::Timestamp) {
        
        // 如果连接管理器里面没有该连接，则这个链接是来历不明的，搜易不能创建信道，并关闭这个不存在的连接
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (connp.get() == nullptr) {
            LOG(Debug) << "打开信道失败, 没有找到对应的链接!" << endl;
            conn->shutdown(); // 关闭这个不存在的连接
        }
        connp->openChannle(req);
    }    

    // 关闭信道
    void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& req, muduo::Timestamp) {
        // 也是先要去找一下是否有对应的连接存在，如果不存在则不用删除
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "删除信道失败，没有找到对应的连接!" << endl;
            conn->shutdown();
        }
        connp->closeChannel(req);
    }  

    // 声明交换机
    void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "声明交换机失败, 没有找到对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "声明交换机失败, 没有找到对应的信道!" << endl;
            return;
        }

        // 通过信道声明交换机
        chn->declareExchange(req);
    }

    // 删除交换机
    void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "删除交换机失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "删除交换机失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->deleteExchange(req);
    }

    // 声明队列
    void onDelareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "声明队列失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "声明队列失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->declareQueue(req);
    }

    // 删除队列
    void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "删除队列失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "删除队列失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->deleteQueue(req);
    }

    // 队列绑定
    void onQueueBinding(const muduo::net::TcpConnectionPtr& conn, const QueueBindingRequestPtr& req, muduo::Timestamp) {

        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "绑定失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "绑定失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->queueBind(req);
    }

    // 队列解除绑定
    void onQueueUnBinding(const muduo::net::TcpConnectionPtr& conn, const QueueUnBindingRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "解除绑定失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "解除绑定失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->queueUnBind(req);
    }

    // 消息发布
    void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublshRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "消息发布失败, 没有找到对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "消息发布失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->basicPublish(req);   
    }    

    // 消息确认
    void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "消息确认失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "消息确认失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->basicAck(req);   
    } 

    // 队列消息订阅
    void onBasicConsumer(const muduo::net::TcpConnectionPtr& conn, const basicConsumerRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "消息订阅失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "消息订阅失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->basicConsumer(req);   
    } 

    // 队列消息取消订阅
    void onBasicConcel(const muduo::net::TcpConnectionPtr& conn, const basicConcelRequestPtr& req, muduo::Timestamp) {
        // 先找到链接，
        Connection::ptr connp = _connection_manager->getConneciont(conn);
        if (conn.get() == nullptr) {
            LOG(Debug) << "取消订阅失败, 没有招待对应的连接!" << endl;
            conn->shutdown();
            return;
        }
        // 再通过连接找到信道
        channel::ptr chn = connp->getChannel(req->cid());
        if (chn.get() == nullptr) {
            LOG(Debug) << "取消订阅失败, 没有找到对应的信道!" << endl;
            return;
        }
        chn->basicCancel(req);  
    } 


    // 未知请求的处理方法
    void OnUnkownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
        LOG(Debug) << "OnUnkownMessage" << endl;
        conn->shutdown();
    }

    void OnConnection(const muduo::net::TcpConnectionPtr& conn) {
        if (conn->connected()) {
            LOG(Info) << "有新链接到来" << endl;
            // 当连接到来的时候，需要去初始化一个连接对象
            _connection_manager->createConnecion(_virtualHost, _consumer_manager, _codec, conn, _threadpool);
        } else {
            // 单链接关闭的时候，需要删除连接对象
            _connection_manager->deleteConnection(conn);
        }
    }

private :
    muduo::net::EventLoop _baseloop;
    muduo::net::TcpServer _server; // 服务器对象
    ProtobufDispatcher _dispatcher; // 请求分发器对象——要想其中注册请求处理函数的
    ProtobufCodecPtr _codec; // protobuf协议处理对象——针对收到的请求数据进行protobuf处理
    unordered_map<string, string> _dict;

    // 数据管理
    VirtualHost::ptr _virtualHost; // 虚拟机数据管理句柄
    consumerManager::ptr _consumer_manager;  // 消费者数据管理句柄
    ConnectionManager::ptr _connection_manager; // 连接管理
    threadpoolPtr _threadpool; // 线程池
};
};
